package com.atguigu.Util;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.util.Properties;

/**
 * @author hjy
 * @create 2023/3/8 11:56
 */
public class MyKafkaUtil {

    private static final String bootStrapServer = "hadoop102:9092,hadoop103:9092,hadoop104:9092";

    public static FlinkKafkaConsumer<String> getFlinkKafkaConsumer(String topic, String groupId) {
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootStrapServer);
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);


        return new FlinkKafkaConsumer<String>(topic, new KafkaDeserializationSchema<String>() {
            @Override
            public boolean isEndOfStream(String nextElement) {
                return false;
            }

            @Override
            public String deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
                if (record == null || record.value() == null) {
                    return null;
                } else {
                    return new String(record.value());
                }

            }

            @Override
            public TypeInformation<String> getProducedType() {
                return BasicTypeInfo.STRING_TYPE_INFO;
            }
        }, props);
    }

    public static FlinkKafkaProducer<String> getKafkaProducer(String topic) {

        return new FlinkKafkaProducer<String>(bootStrapServer,
                topic,
                new SimpleStringSchema());
    }

    public static FlinkKafkaProducer<JSONObject> getKafkaProducer() {
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootStrapServer);

        return new FlinkKafkaProducer<JSONObject>("dwd_default_topic", new KafkaSerializationSchema<JSONObject>() {

            @Override
            public ProducerRecord<byte[], byte[]> serialize(JSONObject element, @Nullable Long timestamp) {
                return new ProducerRecord<>(element.getString("sink_table"), element.getString("data").getBytes());
            }
        }, properties, FlinkKafkaProducer.Semantic.EXACTLY_ONCE);
    }


    public static String getTopicDBDDL(String groupId) {
        return "create table topic_db(\n" +
                "     `database` STRING,\n" +
                "     `table` STRING,\n" +
                "     `type` STRING,\n" +
                "     `data` MAP<STRING,STRING>,\n" +
                "     `old` MAP<STRING,STRING>,\n" +
                "     `pt` AS PROCTIME()\n" +
                ") " + getKafkaDDL(groupId);
    }
    public static String getKafkaDDL(String groupId) {
        return "with(\n" +
                "    'connector' = 'kafka',\n" +
                "    'topic' = 'topic_db',\n" +
                "    'properties.bootstrap.servers' = '" + bootStrapServer + "',\n" +
                "    'properties.group.id' = '" + groupId + "',\n" +
                "    'scan.startup.mode' = 'group-offsets',\n" +
                "    'format' = 'json'\n" +
                ")";
    }
    public static String getKafkaDDL(String topic,String groupId) {
        return "with(\n" +
                "    'connector' = 'kafka',\n" +
                "    'topic' = '"+topic+"',\n" +
                "    'properties.bootstrap.servers' = '" + bootStrapServer + "',\n" +
                "    'properties.group.id' = '" + groupId + "',\n" +
                "    'scan.startup.mode' = 'group-offsets',\n" +
                "    'format' = 'json'\n" +
                ")";
    }

    public static String getKafkaSinkDDL(String topic) {
        return "with(\n" +
                "  'connector'='kafka',\n" +
                "  'topic' = '" + topic + "',\n" +
                "  'properties.bootstrap.servers' = '" + bootStrapServer + "',\n" +
                "  'format'='json'\n" +
                ")";
    }

    public static String getUpsertKafkaDDL(String topic) {
        return "with(\n" +
                "  'connector'='upsert-kafka'," +
                "  'topic' = '" + topic + "'," +
                "  'properties.bootstrap.servers' = '" + bootStrapServer + "'," +
                "  'key.format' = 'json'," +
                "  'value.format' = 'json'" +
                ")";
    }
}
